package cn.lyjuan.first.hadoop.demo;

import cn.lyjuan.base.util.ArrayUtils;
import cn.lyjuan.base.util.JsonUtils;
import cn.lyjuan.base.util.ReflectUtils;
import cn.lyjuan.base.util.StringUtils;
import cn.lyjuan.first.hadoop.util.FileUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class C_001 {
    public static final String HADOOP_URI = "hdfs://192.168.1.200:9000";
    public static final String USER = "root";
    public static final Path LOCAL_PATH = new Path(FileUtil.projectHome() + "/C_001/test.txt");
    public static final Path REMOTE_PATH = new Path("/test.txt");
    public static FileSystem fs;

    @Before
    public void before() throws Exception {
        URI uri = new URI(HADOOP_URI);
        Configuration conf = new Configuration();
        // 单机hadoop情况下，namenode返回datanode的hostname和私有IP地址，只能使用hostname连接，并且需要在本机的/etc/hosts上配置正确的hostname解析
        conf.set("dfs.client.use.datanode.hostname", "true");

        fs = FileSystem.get(uri, conf, USER);
        System.out.println("home ==> " + fs.getHomeDirectory());
    }

    /**
     * 获取集群上所有节点名称
     */
    @Test
    public void nodeList() throws Exception {
        DistributedFileSystem hdfs = (DistributedFileSystem) fs;
        DatanodeInfo[] ds = hdfs.getDataNodeStats();
        int len = ds.length;
        for (int i = 0; i < len; i++) {
            System.out.println("=== " + i + " ===");
            Map<String, Object> map = ReflectUtils.membersToMap(ds[i]);
            for (Map.Entry<String, Object> e : map.entrySet()) {
                System.out.println(e.getKey() + " = " + JsonUtils.to(e.getValue()));
            }
            System.out.println();
            System.out.println();
        }
    }

    /**
     * 获取文件在集群上的位置
     */
    @Test
    public void fileLocation() throws Exception {
        FileStatus status = fs.getFileStatus(REMOTE_PATH);
        BlockLocation[] location = fs.getFileBlockLocations(status, 0, status.getLen());
        int len = location.length;
        for (int i = 0; i < len; i++) {
            String[] hosts = location[i].getHosts();
            System.out.println("block_" + i + " at " + ArrayUtils.join(hosts));
        }
    }

    @Test
    public void read() throws IOException {
        FSDataInputStream in = fs.open(REMOTE_PATH);
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        int len = -1;
        byte[] b = new byte[1024 * 1024];
        while (-1 != (len = in.read(b))) {
            out.write(b, 0, len);
        }

        String content = out.toString(StandardCharsets.UTF_8.name());
        System.out.println("read content ==> " + content);
    }

    @Test
    public void del() throws IOException {
        boolean isDel = fs.deleteOnExit(REMOTE_PATH);
        Assert.assertTrue(isDel);
    }

    @Test
    public void copyToLocalFile() throws IOException {
        File bkFile = new File(LOCAL_PATH.toUri() + ".bk");
        bkFile.deleteOnExit();
        fs.copyToLocalFile(REMOTE_PATH, new Path(bkFile.getAbsolutePath()));
    }

    @Test
    public void copyFromLocalFile() throws IOException {
        fs.copyFromLocalFile(LOCAL_PATH, REMOTE_PATH);
        boolean isCopyOk = fs.exists(REMOTE_PATH);
        System.out.println("copy resp ==> " + isCopyOk);
    }

}